package com.yeming.flink.practice.tablesql

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table, Types}
import org.apache.flink.table.sources.CsvTableSource

object TestCreateTableByDataStreaming {

  def main(args: Array[String]): Unit = {
    //使用flink原生代码创建TableEnvironment
    //先初始化上下文环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    val stream: DataStream[StationLog] = streamEnv.socketTextStream("localhost", 9999).map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    tableEnv.createTemporaryView("t1",stream)

    val table: Table = tableEnv.scan("t1")
    table.printSchema()



  }

}
